package cn.com.ibibj.alignmentService.handler;

import cn.com.ibibj.alignmentService.common.dto.ApiRequestDTO;
import cn.com.ibibj.alignmentService.service.ApiCallerService;
import com.alibaba.fastjson2.JSON;
import io.lettuce.core.RedisClient;
import io.lettuce.core.RedisURI;
import io.lettuce.core.api.reactive.RedisReactiveCommands;
import io.lettuce.core.pubsub.StatefulRedisPubSubConnection;
import io.lettuce.core.pubsub.api.reactive.RedisPubSubReactiveCommands;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;
import reactor.core.publisher.Mono;
import reactor.core.scheduler.Schedulers;

import javax.annotation.Resource;
import java.time.Duration;
import java.util.Map;

@Component
public class RedisRequestHandler {
    private static final String QUEUE_KEY = "ALL_REQUEST";
    private static final String PUB_SUB_CHANNEL = "request_events";
    private final RedisReactiveCommands<String, String> redisCommands;
    private final RedisReactiveCommands<String, String> redisPublishCommands;
    private final RedisPubSubReactiveCommands<String, String> pubSubCommands;
    private final int maxRetries;

    @Resource
    private ApiCallerService apiCallerService;


    @Autowired
    public RedisRequestHandler(@Value("${spring.redis.host}") String redisHost
            , @Value("${spring.redis.port}") Integer port
            , @Value("${redis.max-retries}") Integer maxRetries) {
        this.maxRetries = maxRetries;
        RedisURI redisURI = RedisURI.builder().withHost(redisHost).withPort(port).build();
        // 初始化Lettuce客户端
        RedisClient client = RedisClient.create(redisURI);
        // 获取普通命令连接（线程安全，可共享）
        this.redisCommands = client.connect().reactive();
        // 初始化发布订阅连接，用于监听新请求
        StatefulRedisPubSubConnection<String, String> pubSubConnection = client.connectPubSub();
        this.pubSubCommands = pubSubConnection.reactive();

        // 创建另一个普通命令连接用于发布消息
        this.redisPublishCommands = client.connect().reactive();
        // 订阅频道，监听新请求通知
        setupRequestListener();
    }

    /**
    * 处理用户提交的请求（异步）
    */
    public Mono<String> submitRequest(String ipAndPort, String endpoint, Map<String, Object> params) {
        // 创建请求对象
        ApiRequestDTO request = new ApiRequestDTO(ipAndPort, endpoint, params, maxRetries);
        // 序列化为JSON并添加到队列
        return Mono.just(request)
                .map(JSON::toJSONString)
                .flatMap(serializedReq ->
                        redisCommands.lpush(QUEUE_KEY, serializedReq)
                                .then(Mono.just(request.getId()))
                                .flatMap(id -> redisPublishCommands.publish(PUB_SUB_CHANNEL, "new_request")
                                        .then(Mono.just(id)))
                )
                .subscribeOn(Schedulers.boundedElastic());
    }

    /**
    * 设置请求监听器
    */
    private void setupRequestListener() {
        // 订阅频道，当有新请求时触发处理
        pubSubCommands.subscribe(PUB_SUB_CHANNEL)
                .thenMany(pubSubCommands.observeChannels())
                .subscribe(channelMessage -> {
                    // 有新请求，开始处理队列
                    processQueue();
                });
        // 初始检查队列是否有未处理的请求
        processQueue();
    }

    // 处理队列中的请求
    private void processQueue() {
        // 从队列中获取一个请求（右侧弹出）
        redisCommands.rpop(QUEUE_KEY)
                .flatMap(this::deserializeRequest)
                .flatMap(this::processSingleRequest)
                .subscribe(
                        result -> {
                            // 处理完成后继续处理下一个请求
                            if (result) {
                                processQueue();
                            }
                        },
                        error -> {
                            System.err.println("处理队列错误: " + error.getMessage());
                            // 出错后稍后重试
                            Mono.delay(Duration.ofSeconds(1))
                                    .subscribe(t -> processQueue());
                        }
                );
    }

    // 处理单个请求
    private Mono<Boolean> processSingleRequest(ApiRequestDTO request) {
        System.out.println("处理请求: " + request + ", 重试次数: " + request.getRetryCount());
        return apiCallerService.call(request.getIpAndPort(), request.getEndpoint(), request.getParams())
                .flatMap(success -> {
                    if (success) {
                        System.out.println("请求成功: " + request);
                        return Mono.just(true);
                    } else {
                        // 请求失败，检查是否可以重试
                        request.incrementRetryCount();
                        if (request.canRetry()) {
                            System.out.println("请求失败，将重试: " + request);
                            // 重新加入队列
                            String serialized = JSON.toJSONString(request);
                            return redisCommands.lpush(QUEUE_KEY, serialized)
                                    .then(Mono.just(true));
                        } else {
                            System.out.println("请求失败，已达最大重试次数: " + request);
                            // 可以在这里添加失败处理逻辑，如记录到错误队列
                            return Mono.just(true);
                        }
                    }
                })
                .onErrorResume(error -> {
                    System.err.println("请求处理出错: " + error.getMessage());
                    // 异常情况下的重试处理
                    request.incrementRetryCount();
                    if (request.canRetry()) {
                        String serialized = JSON.toJSONString(request);
                        return redisCommands.lpush(QUEUE_KEY, serialized)
                                .then(Mono.just(true));
                    }
                    return Mono.just(true);
                });
    }

    /**
     * 生成request对象
     */
    private Mono<ApiRequestDTO> deserializeRequest(String data) {
        try {
            ApiRequestDTO request = JSON.parseObject(data, ApiRequestDTO.class);
            return Mono.just(request);
        } catch (Exception e) {
            System.err.println("反序列化请求失败: " + e.getMessage());
            return Mono.error(e);
        }
    }
}
